package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
//KeyedProcessFunction<KEY, IN, OUT>
public class AdjustIsNewProcessFunctionlx extends KeyedProcessFunction<String,String,String>{
   private transient ValueState<String> fistDataState=null;


    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("fist-data",String.class);
        fistDataState = getRuntimeContext().getState(stateDescriptor);

    }

    @Override
    public void processElement(String s,
                               KeyedProcessFunction<String, String, String>.Context context,
                               Collector<String> collector) throws Exception {
        String firstDate = fistDataState.value();

        JSONObject jsonObject = JSON.parseObject(s);
        Long ts = jsonObject.getLong("ts");
        String isNew = jsonObject.getJSONObject("common").getString("is_new");

        if("1".equals(isNew)){
            if(null==firstDate){
                fistDataState.update(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_TIME_FORMAT));
            }else {
                if(!firstDate.equals(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_TIME_FORMAT))){
                    jsonObject.getJSONObject("common").put("is_new","0");
                }
            }
        }else {
            if (null == firstDate){
                fistDataState.update(DateTimeUtil.convertLongToString(ts - 86400000L, DateTimeUtil.DATE_TIME_FORMAT));
            }

        }
        collector.collect(jsonObject.toJSONString());

    }
}
